package org.anno.chunkdb;

import com.google.common.base.Preconditions;
import com.google.common.base.Predicate;
import com.google.common.collect.Iterables;

import org.anno.chunkdb.Chunkdb.Filter;
import org.anno.chunkdb.Metadata.Column.Type;
import org.anno.chunkdb.TypedCellVisitor.CellPredicateVisitor;

/**
 * Filters rows from a given base {@link QueryPlan} based on a {@link Predicate}. The  {@link
 * Chunkdb.Index} for the {@link Predicate} is identified by an int. The first indexes are for the
 * string values of the corresponding columns. Index ids starting from column_count refer to the
 * sub-list of typed columns. For example for columns "name:string, price: double", we have three
 * index IDs, 0 and 1 for the string values of the first two columns, and 2 for the numeric values
 * of the first typed column, price.
 */
public class FilterQueryPlan extends QueryPlan {

  private final QueryPlan basePlan;
  private final Predicate<Host<RowVisitor>> rowPredicate;

  public FilterQueryPlan(QueryPlan basePlan, Predicate<Host<RowVisitor>> rowPredicate) {
    super(basePlan.table, basePlan.sortedColumn);
    this.basePlan = basePlan;
    this.rowPredicate = rowPredicate;
  }

  public FilterQueryPlan(QueryPlan basePlan, Filter filter) {
    this(basePlan, getRowPredicate(filter));
  }

  @Override
  public void accept(final ResponseVisitor visitor) {

    basePlan.accept(new ResponseVisitor() {

      public void visitColumn(int index, String label) {
        visitor.visitColumn(index, label);
      }

      public void visitRows(Iterable<Host<RowVisitor>> rows) {
        visitor.visitRows(Iterables.filter(rows, rowPredicate));
      }
    });
  }

  public static Predicate<Host<RowVisitor>> getRowPredicate(Filter filter) {
    Preconditions.checkArgument((filter.hasDoubleRhs() ? 1 : 0)
                                + (filter.hasLongRhs() ? 1 : 0)
                                + (filter.hasStringRhs() ? 1 : 0) == 1, "filter {0}", filter);
    int column = filter.getColumnId();
    DiffPredicate diffPredicate = DiffPredicate.valueOf(filter.getOp().name());
    if (filter.hasStringRhs()) {
      byte[] rhs = filter.getStringRhsBytes().toByteArray();
      return getStringCellPredicate(diffPredicate, rhs).asRowPredicate(column, Type.STRING);
    }
    if (filter.hasDoubleRhs()) {
      double rhs = filter.getDoubleRhs();
      return getDoubleCellPredicate(diffPredicate, rhs).asRowPredicate(column, Type.NUMBER);
    }
    long rhs = filter.getLongRhs();
    return getLongCellPredicate(diffPredicate, rhs).asRowPredicate(column, Type.DATETIME);
  }

  private static CellPredicateVisitor getDoubleCellPredicate(
      final DiffPredicate diffPredicate, final double rhs) {
    return new CellPredicateVisitor() {
      public void visitDouble(double v) {
        included = diffPredicate.apply(Double.compare(v, rhs));
      }
    };
  }

  private static CellPredicateVisitor getLongCellPredicate(
      final DiffPredicate diffPredicate, final long rhs) {
    return new CellPredicateVisitor() {
      public void visitLong(long v) {
        included = diffPredicate.apply(Long.compare(v, rhs));
      }
    };
  }

  private static CellPredicateVisitor getStringCellPredicate(
      final DiffPredicate diffPredicate, final byte[] rhs) {
    return new CellPredicateVisitor() {
      public void visitString(byte[] buf, int offset, int len) {
        included = diffPredicate.apply(compare(buf, offset, len, rhs));
      }
    };
  }

  static int compare(byte[] buf, int offset, int len, byte[] rhs) {
    int minLen = len < rhs.length ? len : rhs.length;
    for (int r = 0; r < minLen; ) {
      int diff = (buf[offset++] & 255) - (rhs[r++] & 255);
      if (diff != 0) {
        return diff;
      }
    }
    return len - rhs.length;
  }
}
